package com.haier.spark.project.spark

import com.haier.spark.project.constant.Constants
import com.haier.spark.project.util.StringUtils
import org.apache.spark.util.AccumulatorV2

/**
  * session聚合统计Accumulator
  */
class SessionAggrStatAccumulator extends AccumulatorV2[String, String]{

  private var result =
    s"${Constants.SESSION_COUNT}=0|" +
      s"${Constants.TIME_PERIOD_1s_3s}=0|" +
      s"${Constants.TIME_PERIOD_4s_6s}=0|" +
      s"${Constants.TIME_PERIOD_7s_9s}=0|" +
      s"${Constants.TIME_PERIOD_10s_30s}=0|" +
      s"${Constants.TIME_PERIOD_30s_60s}=0|" +
      s"${Constants.TIME_PERIOD_1m_3m}=0|" +
      s"${Constants.TIME_PERIOD_3m_10m}=0|" +
      s"${Constants.TIME_PERIOD_10m_30m}=0|" +
      s"${Constants.TIME_PERIOD_30m}=0|" +
      s"${Constants.STEP_PERIOD_1_3}=0|" +
      s"${Constants.STEP_PERIOD_4_6}=0|" +
      s"${Constants.STEP_PERIOD_7_9}=0|" +
      s"${Constants.STEP_PERIOD_10_30}=0|" +
      s"${Constants.STEP_PERIOD_30_60}=0|" +
      s"${Constants.STEP_PERIOD_60}=0"


  override def isZero: Boolean = {
    true
  }

  override def copy(): AccumulatorV2[String, String] = {
    val sessionAggrStatAccumulator = new SessionAggrStatAccumulator()
    sessionAggrStatAccumulator.result = this.result
    sessionAggrStatAccumulator
  }

  override def reset(): Unit = {
    result=result
  }

  override def add(v: String): Unit = {
    val v1 = result
    val v2 = v
    // 校验：v1为空的话，直接返回v2
    if(null == v1 || v1.eq("")) {
      result = v2
    } else {
      // 使用StringUtils工具类，从v1中，提取v2对应的值，并累加1
      val oldValue: String = StringUtils.getFieldFromConcatString(v1, "\\|", v2)
      if(oldValue != null) {
        // 将范围区间原有的值，累加1
        val newValue = oldValue.toInt + 1
        // 使用StringUtils工具类，将v1中，v2对应的值，设置成新的累加后的值
        result = StringUtils.setFieldInConcatString(v1, "\\|", v2, String.valueOf(newValue))
      } else {
        result = v1
      }
    }
  }

  override def merge(other: AccumulatorV2[String, String]) = other match {
    case map: SessionAggrStatAccumulator =>
      result = other.value
    case _ =>
      throw new UnsupportedOperationException(
        s"Cannot merge ${this.getClass.getName} with ${other.getClass.getName}")
  }

  override def value: String = {
    result
  }
}
